package cn.com.infinitus.uits.service;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {
	
	private static final String topic = "uitsApiToKafka";
	
	private Producer<String, String> producer;

	public KafkaProducerService() {
		Properties props = new Properties();
		props.put(
				"bootstrap.servers",
				"hrtwksvr01.infinitus.com.cn:6667,hrtwksvr02.infinitus.com.cn:6667,hrtwksvr03.infinitus.com.cn:6667");
		props.put("acks", "all");
		// 消息发送失败后的重试次数
		props.put("retries", "0");
		// 消息批处理大小
		props.put("batch.size", "16384");
		props.put("linger.ms", "1");
		props.put("buffer.memory", "33554432");
		props.put("key.serializer",
				"org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer",
				"org.apache.kafka.common.serialization.StringSerializer");
		// 初始化生产端
		/* kafka消息拼接 */
		producer = new KafkaProducer<String, String>(props);
	}

	public void SendMsg(String msg) {
		producer.send(new ProducerRecord<String, String>(topic, msg));
	}
}
